/*
 * Licensed to the Tai Ping Jin Ke
 *
 * Copyright (c) 2022 .
 * All rights reserved.
 * 项目名称：太平金科-基础组件封装层-ElasticSearch-缺省实现
 * 版权说明：本软件属太平金融科技服务(上海)有限公司所有，在未获得太平金融科技服务(上海)有限公司正式授权情况下，任何企业和个人，不能获取、阅读、安装、传播本软件涉及的任何受知识产权保护的内容。
 */
package com.taipingframework.boot.es.facade;

import cn.hutool.core.date.DatePattern;
import cn.hutool.json.JSONConfig;
import cn.hutool.json.JSONUtil;
import com.taipingframework.boot.es.conditions.query.LambdaDeleteWrapper;
import com.taipingframework.boot.es.conditions.query.LambdaQueryWrapper;
import com.taipingframework.boot.es.conditions.query.LambdaUpdateWrapper;
import com.taipingframework.boot.es.metadata.TableInfo;
import com.taipingframework.boot.es.metadata.TableInfoHelper;
import com.taipingframework.boot.web.pojo.PageEntity;
import com.taipingframework.json.JsonExtUtils;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.MultiBucketsAggregation;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.document.Document;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.UpdateQuery;
import org.springframework.data.elasticsearch.core.query.UpdateResponse;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

public final class DefaultElasticAccess implements ElasticAccess {

    private static final JSONConfig JSON_CONFIG = JSONConfig.create().setDateFormat(DatePattern.UTC_MS_PATTERN);
    private ElasticsearchRestTemplate elasticsearchRestTemplate;

    public DefaultElasticAccess(ElasticsearchRestTemplate elasticsearchRestTemplate) {
        this.elasticsearchRestTemplate = elasticsearchRestTemplate;
    }

    /**
     * 获取单条记录
     */
    @Override
    public <T extends Serializable> T queryForObject(String id, Class<? extends T> clazz) {
        return elasticsearchRestTemplate.get(id, clazz);
    }

    /**
     * 查询列表
     */
    @Override
    public <T extends Serializable> List<T> queryForList(LambdaQueryWrapper<T> queryWrapper) {
        Class<T> entityClass = queryWrapper.getEntityClass();
        // 搜索
        NativeSearchQueryBuilder searchQueryBuilder = queryWrapper.getSearchQueryBuilder();
        // 布尔筛选
        searchQueryBuilder.withQuery(queryWrapper.getBoolQueryBuilder());
        // 排序
        queryWrapper.getSortBuilders().forEach(searchQueryBuilder::withSort);

        NativeSearchQuery searchQuery = searchQueryBuilder.build();
        searchQuery.setTrackTotalHits(true);

        return elasticsearchRestTemplate.search(searchQuery, entityClass)
                .getSearchHits()
                .stream()
                .map(SearchHit::getContent).collect(Collectors.toList());

        //System.gc();
        //return list;
    }

    /**
     * 分页查询
     */
    @Override
    public <T extends Object & Serializable> PageEntity<T> queryForPage(LambdaQueryWrapper<T> queryWrapper, int page, int size) {
        Class<T> entityClass = queryWrapper.getEntityClass();
        // 搜索
        NativeSearchQueryBuilder searchQueryBuilder = queryWrapper.getSearchQueryBuilder();
        // 布尔筛选
        searchQueryBuilder.withQuery(queryWrapper.getBoolQueryBuilder());
        // 分页
        searchQueryBuilder.withPageable(PageRequest.of(page - 1, size));
        // 排序
        queryWrapper.getSortBuilders().forEach(searchQueryBuilder::withSort);
        NativeSearchQuery searchQuery = searchQueryBuilder.build();
        searchQuery.setTrackTotalHits(true);

        // 查询结果
        SearchHits<T> searchHits = elasticsearchRestTemplate.search(searchQuery, entityClass);
        // 当前页 数据
        List<T> pageData = searchHits.stream().map(SearchHit::getContent).collect(Collectors.toList());
        // 总数据量
        long total = searchHits.getTotalHits();
        // 取模
        long remainder = total % size;
        // 统计分页总数
        long last = remainder == 0 ? total / size : ((int) (total / size)) + 1;

        //System.gc();
        return new PageEntity<>(page, Long.valueOf(last).intValue(), size, total, pageData);
    }

    /**
     * 删除单条记录
     */
    @Override
    public String deleteForObject(String id, Class<?> clazz) {
        return elasticsearchRestTemplate.delete(id, clazz);
    }

    /**
     * 删除若干条数据
     */
    @Override
    public <T extends Serializable> void deleteForList(LambdaDeleteWrapper<T> deleteWrapper) {
        Class<T> entityClass = deleteWrapper.getEntityClass();
        TableInfo tableInfo = TableInfoHelper.getTableInfo(entityClass);
        // 搜索
        NativeSearchQueryBuilder searchQueryBuilder = deleteWrapper.getSearchQueryBuilder();
        // 布尔筛选
        searchQueryBuilder.withQuery(deleteWrapper.getBoolQueryBuilder());

        NativeSearchQuery query = searchQueryBuilder.build();
        elasticsearchRestTemplate.delete(query, entityClass, IndexCoordinates.of(tableInfo.getTableName()));
    }

    /**
     * 更新单条记录
     */
    @Override
    public <T extends Serializable> boolean updateForObject(LambdaUpdateWrapper<T> updateWrapper) {
        Class<T> entityClass = updateWrapper.getEntityClass();
        TableInfo tableInfo = TableInfoHelper.getTableInfo(entityClass);
        String entityJson = JSONUtil.toJsonStr(updateWrapper.getDocumentMap(), JSON_CONFIG);

        UpdateQuery updateQuery = UpdateQuery.builder(updateWrapper.getDocumentId())
                .withDocument(Document.parse(entityJson))
                .build();
        UpdateResponse updateResponse = elasticsearchRestTemplate.update(updateQuery, IndexCoordinates.of(tableInfo.getTableName()));
        return UpdateResponse.Result.UPDATED == updateResponse.getResult();
    }

    /**
     * 更新单条记录
     */
    @Override
    public <T extends Serializable> boolean updateForObject(T entity) {
        Class<? extends Serializable> entityClass = entity.getClass();
        TableInfo tableInfo = TableInfoHelper.getTableInfo(entityClass);
        String documentId = JsonExtUtils.readObject(entity, tableInfo.getKeyColumn());
        String entityJson = JSONUtil.toJsonStr(entity, JSON_CONFIG);

        UpdateQuery updateQuery = UpdateQuery.builder(documentId)
                .withDocument(Document.parse(entityJson))
                .build();
        UpdateResponse updateResponse = elasticsearchRestTemplate.update(updateQuery, IndexCoordinates.of(tableInfo.getTableName()));
        return UpdateResponse.Result.UPDATED == updateResponse.getResult();
    }

    /**
     * 批量更新
     */
    @Override
    public <T extends Serializable> boolean updateForList(List<T> entityList) {
        Class<? extends Serializable> entityClass = entityList.get(0).getClass();
        TableInfo tableInfo = TableInfoHelper.getTableInfo(entityClass);

        List<UpdateQuery> updateQueryList = new ArrayList<>();
        for (T entity : entityList) {
            String documentId = JsonExtUtils.readObject(entity, tableInfo.getKeyColumn());
            String entityJson = JSONUtil.toJsonStr(entity, JSON_CONFIG);
            updateQueryList.add(UpdateQuery.builder(documentId)
                    .withDocument(Document.parse(entityJson))
                    .build());
        }

        elasticsearchRestTemplate.bulkUpdate(updateQueryList, IndexCoordinates.of(tableInfo.getTableName()));
        return true;
    }

    /**
     * 保存
     */
    @Override
    public <T extends Serializable> T save(T entity) {
        return elasticsearchRestTemplate.save(entity);
    }

    /**
     * 批量新增
     */
    @Override
    public <T extends Serializable> Iterable<T> saveAll(List<T> entityList) {
        return elasticsearchRestTemplate.save(entityList);
    }

    /**
     * 计数
     */
    @Override
    public <T extends Serializable> long count(LambdaQueryWrapper<T> queryWrapper) {
        Class<T> entityClass = queryWrapper.getEntityClass();
        // 搜索
        NativeSearchQueryBuilder searchQueryBuilder = queryWrapper.getSearchQueryBuilder();
        // 布尔筛选
        searchQueryBuilder.withQuery(queryWrapper.getBoolQueryBuilder());

        NativeSearchQuery query = searchQueryBuilder.build();
        return elasticsearchRestTemplate.count(query, entityClass);
    }

    /**
     * 判断是否存在
     */
    @Override
    public boolean exist(String id, Class<?> clazz) {
        return elasticsearchRestTemplate.exists(id, clazz);
    }

    /**
     * 分组聚合统计
     */
    @Override
    public <T extends Serializable> Map<String, List<Map<String, Object>>> aggregate(LambdaQueryWrapper<T> queryWrapper, Class<T> entityClass) {
        entityClass = Objects.isNull(queryWrapper.getEntityClass()) ? entityClass : queryWrapper.getEntityClass();
        TableInfo tableInfo = TableInfoHelper.getTableInfo(entityClass);
        // 搜索
        NativeSearchQueryBuilder searchQueryBuilder = queryWrapper.getSearchQueryBuilder();
        // 布尔筛选
        searchQueryBuilder.withQuery(queryWrapper.getBoolQueryBuilder());
        // 分组
        queryWrapper.getAggregationBuilders().forEach(searchQueryBuilder::addAggregation);
        // 排序
        queryWrapper.getSortBuilders().forEach(searchQueryBuilder::withSort);

        NativeSearchQuery searchQuery = searchQueryBuilder.build();
        searchQuery.setTrackTotalHits(true);

        Map<String, List<Map<String, Object>>> mapResult = new LinkedHashMap<>();
        Aggregations aggregations = elasticsearchRestTemplate.search(searchQuery, entityClass, IndexCoordinates.of(tableInfo.getTableName()))
                .getAggregations();

        if (Objects.isNull(aggregations)) {
            return mapResult;
        }

        for (Aggregation aggregation : aggregations) {
            mapResult.put(aggregation.getName(), readAggregation(aggregation));
        }

        return mapResult;
    }

    /**
     * 读取聚合结果
     */
    private List<Map<String, Object>> readAggregation(Aggregation aggregation) {
        List<Map<String, Object>> list = new ArrayList<>();

        for (MultiBucketsAggregation.Bucket bucket : ((MultiBucketsAggregation) aggregation).getBuckets()) {
            Map<String, Object> bucketMap = new LinkedHashMap<>();
            bucketMap.put(Aggregation.CommonFields.KEY.getPreferredName(), bucket.getKeyAsString());
            bucketMap.put(Aggregation.CommonFields.DOC_COUNT.getPreferredName(), bucket.getDocCount());

            for (Aggregation bucketAggregation : bucket.getAggregations()) {
                bucketMap.put(bucketAggregation.getName(), readAggregation(bucketAggregation));
            }

            list.add(bucketMap);
        }

        return list;
    }

}
